"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : GSC功能模块
Case Name   : ustore表gsc缓存功能验证
Create At   : 2024/01/18
Owner       : @peilinqian
Description :
    1、创建ustore表，并插入1条数据，并查询;
    2、查看syscache占用情况;
    3、增加ustore表索引;
    4、查看syscache占用情况;
    5、删除表;
    6、查看syscache占用情况;
    7、循环进行表删除创建查询操作;
    8、并发循环查询表;
Expect      :
    1、创建ustore表，并插入1条数据，并查询; 操作成功
    2、查看syscache占用情况; 缓存信息正确
    3、增加ustore表索引; 操作成功
    4、查看syscache占用情况; 缓存信息正确
    5、删除表; 操作成功
    6、查看syscache占用情况; 表缓存信息占用记录为空
    7、循环进行表删除创建查询操作; 功能正常
    8、并发循环查询表; 查询结果会间断出现无表报错或者查询正常
History     :
"""
import os
import re
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node

pri_sh = CommonSH('PrimaryDbUser')


class GscTestCase(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)}:start----')
        self.log.info("----this is setup----")
        self.pri_dbuser = Node(node='PrimaryDbUser')
        self.pri_root = Node(node='PrimaryRoot')
        self.tb_name = "tb_gsc0065"
        self.select_thread_num = 10
        self.com = Common()
        self.constant = Constant()

        self.create_sql = f"drop table if exists {self.tb_name};" \
            f"create table {self.tb_name} " \
            f"(id int not null, aa varchar(10) ,id2 int) " \
            f"with (storage_type=ustore);" \
            f"insert into {self.tb_name} values(11, '1a', 11);" \
            f"select * from {self.tb_name};"
        self.select_sql = f"select * from {self.tb_name};"

    def test_main(self):
        step_txt = '----step0:查看数据库状态是否正常;----'
        self.log.info(step_txt)
        status_result = pri_sh.get_db_cluster_status('status')
        self.assertTrue(status_result, '执行失败:' + step_txt)
        step_txt = '----step0:查看enable_global_syscache初始配置值;----'
        self.log.info(step_txt)
        self.init_para1 = self.com.show_param("enable_global_syscache")

        if self.init_para1 == 'off':
            step_txt = '----step1:打开gsc开关; expect:修改成功----'
            self.log.info(step_txt)
            result = pri_sh.execute_gsguc('set',
                                          self.constant.GSGUC_SUCCESS_MSG,
                                          f"enable_global_syscache= on")
            self.assertTrue(result, '执行失败:' + step_txt)

            step_txt = '----step2:重启数据库生效; expect:重启成功----'
            self.log.info(step_txt)
            restart = pri_sh.restart_db_cluster()
            self.assertTrue(restart, '执行失败:' + step_txt)
            self.is_reset_para = True
        else:
            step_txt = '----gsc开关设置正常，跳过step1、step2----'
            self.log.info(step_txt)
            self.is_reset_para = False

        step_txt = '----step1:创建ustore表，并插入1条数据，并查询; expect:操作成功----'
        self.log.info(step_txt)
        result = pri_sh.execut_db_sql(self.create_sql)
        self.log.info(result)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result,
                      "执行失败" + step_txt)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, result,
                      "执行失败" + step_txt)
        self.assertIn('INSERT 0 1', result, "执行失败" + step_txt)
        self.assertIn('1 row', result, "执行失败" + step_txt)

        step_txt = '----step2:查看syscache占用情况; expect:缓存信息正确----'
        self.log.info(step_txt)
        check_sql = f"select * from gs_gsc_table_detail() where " \
            f"relname like '{self.tb_name}%'; "
        check_result = pri_sh.execut_db_sql(check_sql)
        self.log.info(check_result)
        self.assertIn("'id','aa','id2'", check_result, '执行失败:' + step_txt)
        self.assertIn("1 row", check_result, '执行失败:' + step_txt)

        step_txt = '----step3:增加ustore表索引; expect:操作成功----'
        self.log.info(step_txt)
        drop_sql = f"create index on {self.tb_name}(id);" \
            f"select * from {self.tb_name};"
        result = pri_sh.execut_db_sql(drop_sql)
        self.log.info(result)
        self.assertIn(self.constant.CREATE_INDEX_SUCCESS, result,
                      "执行失败" + step_txt)

        step_txt = '----step4:查看syscache占用情况; expect:缓存信息正确----'
        self.log.info(step_txt)
        check_sql = f"select * from gs_gsc_table_detail() where " \
            f"relname like '{self.tb_name}%'; "
        check_result = pri_sh.execut_db_sql(check_sql)
        self.log.info(check_result)
        self.assertIn("'id','aa','id2'", check_result, '执行失败:' + step_txt)
        self.assertIn("2 row", check_result, '执行失败:' + step_txt)

        step_txt = '----step5:删除表; expect:操作成功----'
        self.log.info(step_txt)
        drop_sql = f"drop table if exists {self.tb_name};"
        result = pri_sh.execut_db_sql(drop_sql)
        self.log.info(result)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result,
                      "执行失败" + step_txt)

        step_txt = '----step6:查看syscache占用情况; expect:表缓存信息占用记录为空----'
        self.log.info(step_txt)
        check_result = pri_sh.execut_db_sql(check_sql)
        self.log.info(check_result)
        self.assertIn("0 rows", check_result, '执行失败:' + step_txt)

        step7_txt = '----step7:循环进行表删除创建查询操作; expect:功能正常----'
        self.log.info(step7_txt)
        alter_thread = ComThread(pri_sh.cycle_exec_sql,
                                 args=(self.create_sql, 120, False,
                                       None, True))
        alter_thread.setDaemon(True)
        alter_thread.start()
        time.sleep(0.2)

        step8_txt = '----step8:并发循环查询表;' \
                    'expect:查询结果会间断出现无表报错或者查询正常----'
        self.log.info(step8_txt)
        connect_thread = []
        connect_result = []
        for i in range(self.select_thread_num):
            connect_thread.append(ComThread(pri_sh.cycle_exec_sql, args=(
                self.select_sql, 120, False, None, True)))
            connect_thread[i].setDaemon(True)
            connect_thread[i].start()

        self.log.info('----获取步骤7执行结果----')
        alter_thread.join(600)
        exec_result = alter_thread.get_result()
        pattern = r'DROP TABLE\s+.*CREATE TABLE\s+INSERT 0 1\s+.*\(1 row\).*?'
        for i in set(exec_result):
            self.log.info(i)
            if not re.search(pattern, i, re.S | re.M):
                self.log.info(i)
            self.assertIsNotNone(re.search(pattern, i, re.S | re.M),
                                 '执行失败:' + step7_txt)

        self.log.info('----获取步骤8执行结果----')
        pattern = r'ERROR:  relation ".*" does not exist on dn_6001'
        for i in range(self.select_thread_num):
            connect_thread[i].join(600)
            connect_result.extend(connect_thread[i].get_result())
        self.log.info('执行总次数' + str(len(connect_result)))
        set_msg = set(connect_result)
        self.log.info(set_msg)
        self.assertLessEqual(len(set_msg), 3, '执行失败:' + step8_txt)
        if len(set_msg) == 1:
            self.assertEqual(str(set_msg).count('id | aa | id2'), 1,
                             '执行失败:' + step8_txt)
        if len(set_msg) == 2:
            if str(set_msg).count('id | aa | id2') < 2:
                self.assertIsNotNone(
                    re.search(pattern, str(set_msg), re.S | re.M),
                    '执行失败:' + step8_txt)
            else:
                self.assertEqual(str(set_msg).count('id | aa | id2'),
                                 2, '执行失败:' + step8_txt)
        if len(set_msg) == 3:
            self.assertEqual(str(set_msg).count('id | aa | id2'), 2,
                             '执行失败:' + step8_txt)
            self.assertIsNotNone(
                re.search(pattern, str(set_msg), re.S | re.M),
                '执行失败:' + step8_txt)

    def tearDown(self):
        self.log.info('----this is tearDown----')
        step11_txt = '----清理ustore表; expect:清理成功----'
        self.log.info(step11_txt)
        drop_cmd = f"drop table if exists {self.tb_name};"
        drop_result = pri_sh.execut_db_sql(drop_cmd)
        self.log.info(drop_result)

        if self.is_reset_para:
            self.step12_txt = '----恢复参数为初始值并查询; expect:设置成功----'
            self.log.info(self.step12_txt)
            self.msg1 = pri_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f"enable_global_syscache ={self.init_para1}")
            self.step13_txt = '----重启数据库生效; expect:重启成功----'
            self.log.info(self.step13_txt)
            self.restart = pri_sh.restart_db_cluster()
            self.step4_txt = '----查询数据库状态; expect:状态正常----'
            self.log.info(self.step4_txt)
            self.status_result = pri_sh.get_db_cluster_status('status')

        step_txt = '----teardown断言----'
        self.log.info(f'----{os.path.basename(__file__)}:end----')
        self.log.info(step_txt)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_result,
                      "执行失败" + step11_txt)
        if self.is_reset_para:
            self.assertTrue(self.msg1, '执行失败:' + self.step12_txt)
            self.assertTrue(self.restart, '执行失败:' + self.step13_txt)
            self.assertTrue(self.status_result, '执行失败:' + self.step4_txt)
